package ext.opensource.netty.client.mqtt.protocol;

import ext.opensource.netty.client.core.ClientProcess;
import ext.opensource.netty.client.mqtt.api.MqttConsumerProcess;
import ext.opensource.netty.client.mqtt.api.MqttProducerProcess;
import ext.opensource.netty.client.mqtt.common.MessageData;
import ext.opensource.netty.client.mqtt.common.MessageStatus;
import ext.opensource.netty.common.NettyLog;
import ext.opensource.netty.common.exception.LoginException;
import ext.opensource.netty.common.utils.ByteBufUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;

public class ClientProtocolProcess {
	private final ClientProcess clientProcess;
	private final MqttConsumerProcess consumerProcess;
	private final MqttProducerProcess producerProcess;

	public ClientProtocolProcess(final ClientProcess clientProcess, final MqttConsumerProcess consumerProcess,
			final MqttProducerProcess producerProcess) {
		this.clientProcess = clientProcess;
		this.consumerProcess = consumerProcess;
		this.producerProcess = producerProcess;
	}

	/**
	 * B - S, B - P
	 * 
	 * @param channel
	 * @param msg
	 */
	public void processConnectBack(final Channel channel, final MqttConnAckMessage msg) {
		final MqttConnAckVariableHeader mqttConnAckVariableHeader = msg.variableHeader();
		String sErrorMsg = "";
		switch (mqttConnAckVariableHeader.connectReturnCode()) {
			case CONNECTION_ACCEPTED:
				clientProcess.loginFinish(true, null);
				return;
			case CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD:
				sErrorMsg = "用户名密码错误";
				break;
			case CONNECTION_REFUSED_IDENTIFIER_REJECTED:
				sErrorMsg = "clientId不允许链接";
				break;
			case CONNECTION_REFUSED_SERVER_UNAVAILABLE:
				sErrorMsg = "服务不可用";
				break;
			case CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION:
				sErrorMsg = "mqtt 版本不可用";
				break;
			case CONNECTION_REFUSED_NOT_AUTHORIZED:
				sErrorMsg = "未授权登录";
				break;
			default:
				break;
		}

		clientProcess.loginFinish(false, new LoginException(sErrorMsg));
		channel.close();
	}

	/**
	 * B - P (Qos1)
	 * 
	 * @param channel
	 * @param mqttMessage
	 */
	public void processPubAck(final Channel channel, final MqttMessage mqttMessage) {
		final MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		final int messageId = messageIdVariableHeader.messageId();

		producerProcess.processPubAck(messageId);
		producerProcess.delMessage(messageId);
	}

	/**
	 * B- P(Qos2)
	 * 
	 * @param channel
	 * @param mqttMessage
	 */
	public void processPubRec(final Channel channel, final MqttMessage mqttMessage) {
		final MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		final int messageId = messageIdVariableHeader.messageId();

		producerProcess.processPubRec(messageId);
		producerProcess.sendPubRel(messageId);
	}

	/**
	 * B - P (Qos2)
	 * 
	 * @param channel
	 * @param mqttMessage
	 */
	public void processPubComp(final Channel channel, final MqttMessage mqttMessage) {
		final MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		final int messageId = messageIdVariableHeader.messageId();

		producerProcess.processPubComp(messageId);
		producerProcess.delMessage(messageId);
	}

	/**
	 * B - S(Qos2)
	 * 
	 * @param channel
	 * @param mqttMessage
	 */
	public void processPubRel(final Channel channel, final MqttMessage mqttMessage) {
		final MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		final int messageId = messageIdVariableHeader.messageId();
		this.consumerProcess.processPubRel(messageId);
		this.consumerProcess.sendPubCompMessage(messageId);
		this.consumerProcess.delMesage(messageId);
	}

	/**
	 * B - S(Qos0, Qos1, Qos2)
	 * 
	 * @param channel
	 * @param mqttMessage
	 */
	public void processPublish(final Channel channel, final MqttPublishMessage mqttMessage) {
		final MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
		final MqttPublishVariableHeader mqttPublishVariableHeader = mqttMessage.variableHeader();
		final ByteBuf payload = mqttMessage.payload();
		final String topciName = mqttPublishVariableHeader.topicName();
		final MqttQoS qosLevel = mqttFixedHeader.qosLevel();
		final int messageId = mqttPublishVariableHeader.packetId();
		final byte[] bytes = ByteBufUtil.copyByteBuf(payload);

		final MessageData recviceMessage = MessageData.builder().topic(topciName).qos(qosLevel.value())
				.messageId(messageId).payload(bytes).status(MessageStatus.PUB).dup(mqttFixedHeader.isDup())
				.retained(mqttFixedHeader.isRetain()).build();

		if (qosLevel == MqttQoS.EXACTLY_ONCE) {
			this.consumerProcess.saveMesage(recviceMessage);
		}

		this.consumerProcess.processPublish(recviceMessage);

		switch (qosLevel) {
			case AT_MOST_ONCE:
				break;
			case AT_LEAST_ONCE:
				this.consumerProcess.sendPubAckMessage(messageId);
				break;
			case EXACTLY_ONCE:
				this.consumerProcess.sendPubRecMessage(messageId);
				break;
			default:
				break;
		}
	}

	/**
	 * B - P
	 * 
	 * @param channel
	 * @param mqttMessage
	 */
	public void processSubAck(final Channel channel, final MqttSubAckMessage mqttMessage) {
		final int messageId = mqttMessage.variableHeader().messageId();
		this.consumerProcess.processSubAck(messageId);
	}

	/**
	 * B - S
	 * 
	 * @param channel
	 * @param mqttMessage
	 */
	public void processUnSubBack(final Channel channel, final MqttMessage mqttMessage) {
		int messageId;
		if (mqttMessage instanceof MqttUnsubAckMessage) {
			final MqttUnsubAckMessage mqttUnsubAckMessage = (MqttUnsubAckMessage) mqttMessage;
			messageId = mqttUnsubAckMessage.variableHeader().messageId();
		} else {
			final MqttMessageIdVariableHeader o = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
			messageId = o.messageId();
			NettyLog.error("not UnsubAckMessage:{}", messageId);
		}
		this.consumerProcess.processUnSubBack(messageId);
	}
}
